package com.example.demo.service;

/**
 * @author vaintale
 * @date 2025/7/7
 */

import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.ApplicationRunner;
import org.springframework.context.annotation.Bean;
import org.springframework.data.redis.connection.stream.*;
import org.springframework.data.redis.core.StreamOperations;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Service;

import java.time.Duration;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicLong;

@Service
@Slf4j
public class RedisStreamNodeConsumer {

    private final String STREAM_KEY = "redis-stream-02";
    private final String GROUP = "group-node";
    private final String CONSUMER = "consumer-1";
    private final StreamOperations<String, String, String> streamOps;
    private final ObjectMapper objectMapper = new ObjectMapper();
    private Set<String> expectedSources = new HashSet<>();

    private final StringRedisTemplate redisTemplate;
    // index -> source -> value
    private final Map<Integer, Map<String, String>> batchMap = new ConcurrentHashMap<>();
    private final Map<Integer, Map<String, String>> batchMap2 = new ConcurrentHashMap<>();
    private volatile int nextIndex = 1;
    private final AtomicLong intervalMillis = new AtomicLong(1000);


    public RedisStreamNodeConsumer(StringRedisTemplate redisTemplate) {
        this.redisTemplate = redisTemplate;
        this.streamOps = redisTemplate.opsForStream();
        try {
            streamOps.createGroup(STREAM_KEY, ReadOffset.latest(), GROUP);
        } catch (Exception ignore) {}

    }
    @Bean
    public ApplicationRunner runnerNode() {
        return args -> {
            while (true) {
                poll();
                Thread.sleep(1000); // 固定1秒轮询
            }
        };
    }
    public void poll() {
        // 消费Stream
        List<MapRecord<String, String, String>> messages = streamOps.read(
                Consumer.from(GROUP, CONSUMER),
                StreamReadOptions.empty().count(100).block(Duration.ofSeconds(1)),
                StreamOffset.create(STREAM_KEY, ReadOffset.lastConsumed())
        );
        if (messages != null) {
            for (MapRecord<String, String, String> msg : messages) {
                String source = msg.getValue().get("source");
                int index = Integer.parseInt(msg.getValue().get("index"));
                String value = msg.getValue().get("valueGd");
                String value2 = msg.getValue().get("valueXj");
                batchMap.computeIfAbsent(index, k -> new HashMap<>()).put(source, value);
                batchMap2.computeIfAbsent(index, k -> new HashMap<>()).put(source, value2);
                streamOps.acknowledge(STREAM_KEY, GROUP, msg.getId());
               // streamOps.delete(STREAM_KEY, msg.getId());
            }
        }
        // 顺序判断
        while (true) {
//            // 动态从Redis获取本轮应收source
//            Set<String> expectedSources = redisTemplate.opsForSet()
//                    .members(EXPECTED_SOURCES_PREFIX);
            if (expectedSources == null || expectedSources.isEmpty()) {
                // 本轮尚未注册expected-sources，等待生产端写入
                break;
            }
            // 收齐所有expectedSources才处理
            if (batchMap.containsKey(nextIndex) &&
                    batchMap.get(nextIndex).keySet().containsAll(expectedSources)&&batchMap2.containsKey(nextIndex) &&
                    batchMap2.get(nextIndex).keySet().containsAll(expectedSources)) {
                Map<String, String> batch = batchMap.get(nextIndex);
                Map<String, String> batch2 = batchMap2.get(nextIndex);
                processBatch(nextIndex, batch, expectedSources);
                processBatch(nextIndex, batch2, expectedSources);
                batchMap.remove(nextIndex);
                batchMap2.remove(nextIndex);
                // 处理完可删掉expected-sources:{index}
                //   redisTemplate.delete(EXPECTED_SOURCES_PREFIX + nextIndex);
                nextIndex++;
            } else {
                break;
            }
        }
    }

    /**
     * 只对传入的expectedSources做处理
     */
    private void processBatch(int index, Map<String, String> batch, Set<String> expectedSources) {
        System.out.println("==== Index: " + index + " ====");
        List<String> metrics = List.of("Fdelay", "Fthroughput", "Fjitter");
        for (String source : expectedSources) {
            String valueJson = batch.get(source);
            if (valueJson == null) {
                System.out.printf("source: %s (no data)\n", source);
                continue;
            }
            try {
                Map<String, Object> value = objectMapper.readValue(valueJson, new TypeReference<>() {
                });
                System.out.println("source: " + source);
                for (String metric : metrics) {
                    Object metricObj = value.get(metric);
                    if (metricObj == null) {
                        System.out.printf("  %s: null\n", metric);
                    } else if (metricObj instanceof Map) {
                        Map<String, Object> metricMap = (Map<String, Object>) metricObj;
                        double sum = 0;
                        int count = 0;
                        for (Object v : metricMap.values()) {
                            if (v instanceof Number) {
                                sum += ((Number) v).doubleValue();
                                count++;
                            }
                        }
                        if (count > 0) {
                            System.out.printf("  %s: %.8f\n", metric, sum / count);
                        } else {
                            System.out.printf("  %s: null\n", metric);
                        }
                    } else {
                        System.out.printf("  %s: null\n", metric);
                    }
                }
            } catch (Exception e) {
                System.out.printf("source: %s (parse error)\n", source);
                e.printStackTrace();
            }
        }
    }

    public String getIntervalMillis() {
        return String.valueOf(intervalMillis.get());
    }

    public void setIntervalMillis(long millis) {
        this.intervalMillis.set(millis);
    }
    /**
     * 重置消费者状态到初始
     */
    public synchronized void resetConsumer(Set<String> data) {
        batchMap.clear();
        nextIndex = 1;
        expectedSources = data;

        log.info("Consumer has been reset to initial state.");
    }


}
